package com.threez.stream.ibm.binder;

import lombok.extern.slf4j.Slf4j;
import org.springframework.cloud.stream.binder.Binder;
import org.springframework.cloud.stream.binder.ConsumerProperties;
import org.springframework.cloud.stream.binder.ProducerProperties;
import org.springframework.jms.core.JmsTemplate;
import org.springframework.messaging.MessageChannel;
import org.springframework.messaging.SubscribableChannel;
import org.springframework.messaging.support.GenericMessage;
import org.springframework.util.Assert;

import javax.jms.*;
import java.util.function.Function;

@Slf4j
public abstract class JmsBinderAbstract implements Binder<MessageChannel, ConsumerProperties, ProducerProperties> {
    /**
     * 消费者监听
     *
     * @param jmsTemplate       jms消费模板
     * @param inboundBindTarget 监听消息存放到channel通道
     * @param destination       获取消费目标
     */
    public void listenerConsumer(JmsTemplate jmsTemplate, MessageChannel inboundBindTarget, Function<Session, Destination> destination) {
        ConnectionFactory connectionFactory = jmsTemplate.getConnectionFactory();
        try {
            // 创造 JMS 链接
            Connection connection = connectionFactory.createConnection();
            // 启动连接
            connection.start();
            // 创建会话 Session
            Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);

            // 创建消息消费者
            MessageConsumer messageConsumer = session.createConsumer(destination.apply(session));
            messageConsumer.setMessageListener(message -> {
                log.debug("开始监听的message类型：{} ", message.getClass().getName());
                try {
                    if (message instanceof ObjectMessage) { //一般是activeMQ
                        ObjectMessage objectMessage = (ObjectMessage) message;
                        Object object = objectMessage.getObject();
                        inboundBindTarget.send(new GenericMessage<Object>(object));
                    } else if (message instanceof TextMessage) {
                        String msgStr = ((TextMessage) message).getText();
                        inboundBindTarget.send(new GenericMessage<Object>(msgStr));
                    } else if (message instanceof BytesMessage) {
                        BytesMessage byteMessage = (BytesMessage) message;
                        byte[] byteArr = new byte[(int) byteMessage.getBodyLength()];
                        for (int i = 0; i < (int) byteMessage.getBodyLength(); i++) {
                            byteArr[i] = byteMessage.readByte();
                        }
                        String msgStr = new String(byteArr);
                        inboundBindTarget.send(new GenericMessage<Object>(msgStr));
                    } else {
                        log.info("监听IBM MQ的消息类型不存在, {}", message.getClass().getName());
                    }

                } catch (Exception e) {
                    log.error("jms监听消息异常1：", e);
                }

            });
        } catch (Exception e) {
            log.error("jms监听消息异常2：", e);
        }
    }

    /**
     * 生产者
     *
     * @param jmsTemplate
     * @param topic
     * @param outboundBindTarget
     */
    public void sendProducer(JmsTemplate jmsTemplate, String topic, MessageChannel outboundBindTarget) {
        // MessageChannel 必须是 SubscribableChannel 类型
        Assert.isInstanceOf(SubscribableChannel.class, outboundBindTarget, "Binding is supported only for SubscribableChannel instances");
        //强转为 SubscribableChannel类型
        SubscribableChannel subscribableChannel = (SubscribableChannel) outboundBindTarget;
        subscribableChannel.subscribe(message -> {
            Object messageBody = message.getPayload();
            jmsTemplate.convertAndSend(topic, messageBody);
        });
    }
}
